package cn.home.bootstrap.workqueues;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;

/**
 *一个生产者，多个消费者，消费者
 *  接收端和场景1不同点：
 *
 * 1、使用“task_queue”声明消息队列，并使消息队列durable
 * 2、在使用channel.basicConsume接收消息时使autoAck为false，即不自动会发ack，由channel.basicAck()在消息处理完成后发送消息。
 * 3、使用了channel.basicQos(1)保证在接收端一个消息没有处理完时不会接收另一个消息，即接收端发送了ack后才会接收下一个消息。
 * 在这种情况下发送端会尝试把消息发送给下一个not busy的接收端。
 */
@Slf4j
public class Worker {
  private final static String IP = "127.0.0.1";
  private final static int  PORT = 5672;
  private static final String TASK_QUEUE_NAME = "task_queue";

  public static void main(String[] argv) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost(IP);
    factory.setPort(PORT);
    final Connection connection = factory.newConnection();
    final Channel channel = connection.createChannel();
    //队列持久化
    channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
    //配置prefetchCount，在消费者，没有完全处理消息结束（没有ack），broker不会推送消息到消费者
    channel.basicQos(1);
    final Consumer consumer = new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        String message = new String(body, "UTF-8");
        log.info(" [x] Received '" + message + "'");
        try {
          doWork(message);
        } finally {
          log.info(" [x] Done");
          //处理完消息，返回ack消息给broker
          channel.basicAck(envelope.getDeliveryTag(), false);
        }
      }
    };
    //设置自动回复为false，及手动回复，注意当消费者处理消息消耗过多的时间，可能导致broker缓存过多的消息，导致broke内存爆掉
    channel.basicConsume(TASK_QUEUE_NAME, false, consumer);
  }

  /**
   * 模拟消费者，处理消息耗时
   * @param task
   */
  private static void doWork(String task) {
    for (char ch : task.toCharArray()) {
      if (ch == '.') {
        try {
          log.info("consumer messgae:{}",task);
          Thread.sleep(10000);
        } catch (InterruptedException _ignored) {
          Thread.currentThread().interrupt();
        }
      }
    }
  }
}

